package org.nbict.iot.trash;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.ibatis.io.Resources;
import org.apache.ibatis.session.SqlSession;
import org.apache.ibatis.session.SqlSessionFactory;
import org.apache.ibatis.session.SqlSessionFactoryBuilder;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.BasicOutputCollector;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseBasicBolt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;
import org.nbict.iot.protocol.config.ProtocolMapper;
import org.nbict.iot.protocol.context.FieldTypeContext;
import org.nbict.iot.protocol.context.ParserContext;
import org.nbict.iot.protocol.type.ProtocolField;
import org.nbict.iot.protocol.util.BytesOperator;

import java.io.IOException;
import java.io.InputStream;
import java.util.Map;

/**
 * Created by songseven on 18/6/7.
 */
public class ParserBolt extends BaseBasicBolt {

    private ProtocolField rootField;
    private final static byte[] tag = {0x01};
    private final static byte[] patch = {0x23};
    private int count = 0;

    public void prepare(Map stormConf, TopologyContext context) {
        String resource = "mybatis-config.xml";
        InputStream inputStream = null;
        SqlSession session = null;
        try {
            inputStream = Resources.getResourceAsStream(resource);
            SqlSessionFactory sqlSessionFactory = new
                    SqlSessionFactoryBuilder().build(inputStream);
            session = sqlSessionFactory.openSession();
            ProtocolMapper mapper = session.getMapper(ProtocolMapper.class);
            rootField = mapper.getProtocolFieldById(120);
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            session.close();
        }
    }

    @Override
    public void execute(Tuple input, BasicOutputCollector
            collector) {

        String topic = (String)input.getValue(0);// 打印主题
        Long offset = (Long) input.getValue(2); //offset

        byte[] raw = (byte[]) input.getValue(4);
        byte[] checksum = BytesOperator.splitBytes(raw, 0, 1);

        switch (checksum[0]) {
            case (byte)0x01:
                raw = BytesOperator.splitFrontBytes(raw, 1);
            case (byte)0x23: //打补丁
                try {
                    ParserContext context = new FieldTypeContext(raw, rootField);
                    JSON re = (JSON) context.type().process(context);
                    ((JSONObject)re).put("company_id", topic);
                    ((JSONObject)re).put("offset", offset);
                    System.out.println(Thread.currentThread().getName() + " : count: " + count++ + " ==> offset: " + offset); //debug

                    //System.out.println(re.toJSONString());
                    collector.emit("S0", new Values(re)); //debug
                } catch (Exception e) {
                    //TODO 错误流
                }
                break;
            default:
                //TODO 错误流
        }

        //0x01为校验正确

    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declareStream("S0", new Fields("parsed")); //correct
        declarer.declareStream("S1", new Fields("raw")); //error
//        declarer.declare(new Fields("body"));
    }
}
